{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# AirFlow-Tutorial\n",
    "\n",
    "- 入门教程，https://my.oschina.net/u/2306127/blog/1843515\n",
    "- 使用文档，https://airflow.incubator.apache.org/tutorial.html\n",
    "- 项目源码，https://github.com/apache/incubator-airflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Help on class DAG in module airflow.models:\n",
      "\n",
      "class DAG(airflow.dag.base_dag.BaseDag, airflow.utils.logging.LoggingMixin)\n",
      " |  A dag (directed acyclic graph) is a collection of tasks with directional\n",
      " |  dependencies. A dag also has a schedule, a start end an end date\n",
      " |  (optional). For each schedule, (say daily or hourly), the DAG needs to run\n",
      " |  each individual tasks as their dependencies are met. Certain tasks have\n",
      " |  the property of depending on their own past, meaning that they can't run\n",
      " |  until their previous schedule (and upstream tasks) are completed.\n",
      " |  \n",
      " |  DAGs essentially act as namespaces for tasks. A task_id can only be\n",
      " |  added once to a DAG.\n",
      " |  \n",
      " |  :param dag_id: The id of the DAG\n",
      " |  :type dag_id: string\n",
      " |  :param description: The description for the DAG to e.g. be shown on the webserver\n",
      " |  :type description: string\n",
      " |  :param schedule_interval: Defines how often that DAG runs, this\n",
      " |      timedelta object gets added to your latest task instance's\n",
      " |      execution_date to figure out the next schedule\n",
      " |  :type schedule_interval: datetime.timedelta or\n",
      " |      dateutil.relativedelta.relativedelta or str that acts as a cron\n",
      " |      expression\n",
      " |  :param start_date: The timestamp from which the scheduler will\n",
      " |      attempt to backfill\n",
      " |  :type start_date: datetime.datetime\n",
      " |  :param end_date: A date beyond which your DAG won't run, leave to None\n",
      " |      for open ended scheduling\n",
      " |  :type end_date: datetime.datetime\n",
      " |  :param template_searchpath: This list of folders (non relative)\n",
      " |      defines where jinja will look for your templates. Order matters.\n",
      " |      Note that jinja/airflow includes the path of your DAG file by\n",
      " |      default\n",
      " |  :type template_searchpath: string or list of stings\n",
      " |  :param user_defined_macros: a dictionary of macros that will be exposed\n",
      " |      in your jinja templates. For example, passing ``dict(foo='bar')``\n",
      " |      to this argument allows you to ``{{ foo }}`` in all jinja\n",
      " |      templates related to this DAG. Note that you can pass any\n",
      " |      type of object here.\n",
      " |  :type user_defined_macros: dict\n",
      " |  :param default_args: A dictionary of default parameters to be used\n",
      " |      as constructor keyword parameters when initialising operators.\n",
      " |      Note that operators have the same hook, and precede those defined\n",
      " |      here, meaning that if your dict contains `'depends_on_past': True`\n",
      " |      here and `'depends_on_past': False` in the operator's call\n",
      " |      `default_args`, the actual value will be `False`.\n",
      " |  :type default_args: dict\n",
      " |  :param params: a dictionary of DAG level parameters that are made\n",
      " |      accessible in templates, namespaced under `params`. These\n",
      " |      params can be overridden at the task level.\n",
      " |  :type params: dict\n",
      " |  :param concurrency: the number of task instances allowed to run\n",
      " |      concurrently\n",
      " |  :type concurrency: int\n",
      " |  :param max_active_runs: maximum number of active DAG runs, beyond this\n",
      " |      number of DAG runs in a running state, the scheduler won't create\n",
      " |      new active DAG runs\n",
      " |  :type max_active_runs: int\n",
      " |  :param dagrun_timeout: specify how long a DagRun should be up before\n",
      " |      timing out / failing, so that new DagRuns can be created\n",
      " |  :type dagrun_timeout: datetime.timedelta\n",
      " |  :param sla_miss_callback: specify a function to call when reporting SLA\n",
      " |      timeouts.\n",
      " |  :type sla_miss_callback: types.FunctionType\n",
      " |  :param orientation: Specify DAG orientation in graph view (LR, TB, RL, BT)\n",
      " |  :type orientation: string\n",
      " |  :param catchup: Perform scheduler catchup (or only run latest)? Defaults to True\n",
      " |  \"type catchup: bool\"\n",
      " |  \n",
      " |  Method resolution order:\n",
      " |      DAG\n",
      " |      airflow.dag.base_dag.BaseDag\n",
      " |      airflow.utils.logging.LoggingMixin\n",
      " |      builtins.object\n",
      " |  \n",
      " |  Methods defined here:\n",
      " |  \n",
      " |  __deepcopy__(self, memo)\n",
      " |  \n",
      " |  __enter__(self)\n",
      " |  \n",
      " |  __eq__(self, other)\n",
      " |      Return self==value.\n",
      " |  \n",
      " |  __exit__(self, _type, _value, _tb)\n",
      " |  \n",
      " |  __ge__(self, other, NotImplemented=NotImplemented)\n",
      " |      Return a >= b.  Computed by @total_ordering from (not a < b).\n",
      " |  \n",
      " |  __gt__(self, other, NotImplemented=NotImplemented)\n",
      " |      Return a > b.  Computed by @total_ordering from (not a < b) and (a != b).\n",
      " |  \n",
      " |  __hash__(self)\n",
      " |      Return hash(self).\n",
      " |  \n",
      " |  __init__(self, dag_id, description='', schedule_interval=datetime.timedelta(1), start_date=None, end_date=None, full_filepath=None, template_searchpath=None, user_defined_macros=None, default_args=None, concurrency=16, max_active_runs=16, dagrun_timeout=None, sla_miss_callback=None, orientation='LR', catchup=True, params=None)\n",
      " |      Initialize self.  See help(type(self)) for accurate signature.\n",
      " |  \n",
      " |  __le__(self, other, NotImplemented=NotImplemented)\n",
      " |      Return a <= b.  Computed by @total_ordering from (a < b) or (a == b).\n",
      " |  \n",
      " |  __lt__(self, other)\n",
      " |      Return self<value.\n",
      " |  \n",
      " |  __ne__(self, other)\n",
      " |      Return self!=value.\n",
      " |  \n",
      " |  __repr__(self)\n",
      " |      Return repr(self).\n",
      " |  \n",
      " |  add_task(self, task)\n",
      " |      Add a task to the DAG\n",
      " |      \n",
      " |      :param task: the task you want to add\n",
      " |      :type task: task\n",
      " |  \n",
      " |  add_tasks(self, tasks)\n",
      " |      Add a list of tasks to the DAG\n",
      " |      \n",
      " |      :param task: a lit of tasks you want to add\n",
      " |      :type task: list of tasks\n",
      " |  \n",
      " |  clear(self, start_date=None, end_date=None, only_failed=False, only_running=False, confirm_prompt=False, include_subdags=True, reset_dag_runs=True, dry_run=False)\n",
      " |      Clears a set of task instances associated with the current dag for\n",
      " |      a specified date range.\n",
      " |  \n",
      " |  cli(self)\n",
      " |      Exposes a CLI specific to this DAG\n",
      " |  \n",
      " |  crawl_for_tasks(objects)\n",
      " |      Typically called at the end of a script by passing globals() as a\n",
      " |      parameter. This allows to not explicitly add every single task to the\n",
      " |      dag explicitly.\n",
      " |  \n",
      " |  create_dagrun(self, run_id, state, execution_date=None, start_date=None, external_trigger=False, conf=None, session=None)\n",
      " |      Creates a dag run from this dag including the tasks associated with this dag.\n",
      " |      Returns the dag run.\n",
      " |      \n",
      " |      :param run_id: defines the the run id for this dag run\n",
      " |      :type run_id: string\n",
      " |      :param execution_date: the execution date of this dag run\n",
      " |      :type execution_date: datetime\n",
      " |      :param state: the state of the dag run\n",
      " |      :type state: State\n",
      " |      :param start_date: the date this dag run should be evaluated\n",
      " |      :type start_date: datetime\n",
      " |      :param external_trigger: whether this dag run is externally triggered\n",
      " |      :type external_trigger: bool\n",
      " |      :param session: database session\n",
      " |      :type session: Session\n",
      " |  \n",
      " |  date_range(self, start_date, num=None, end_date=datetime.datetime(2018, 7, 21, 11, 51, 12, 893119))\n",
      " |  \n",
      " |  db_merge(self)\n",
      " |  \n",
      " |  following_schedule(self, dttm)\n",
      " |  \n",
      " |  get_active_runs(self, session=None)\n",
      " |      Returns a list of \"running\" tasks\n",
      " |      :param session:\n",
      " |      :return: List of execution dates\n",
      " |  \n",
      " |  get_dagrun(self, execution_date, session=None)\n",
      " |      Returns the dag run for a given execution date if it exists, otherwise\n",
      " |      none.\n",
      " |      :param execution_date: The execution date of the DagRun to find.\n",
      " |      :param session:\n",
      " |      :return: The DagRun if found, otherwise None.\n",
      " |  \n",
      " |  get_last_dagrun(self, session=None, include_externally_triggered=False)\n",
      " |      Returns the last dag run for this dag, None if there was none.\n",
      " |      Last dag run can be any type of run eg. scheduled or backfilled.\n",
      " |      Overriden DagRuns are ignored\n",
      " |  \n",
      " |  get_task(self, task_id)\n",
      " |  \n",
      " |  get_task_instances(self, session, start_date=None, end_date=None, state=None)\n",
      " |  \n",
      " |  get_template_env(self)\n",
      " |      Returns a jinja2 Environment while taking into account the DAGs\n",
      " |      template_searchpath and user_defined_macros\n",
      " |  \n",
      " |  has_task(self, task_id)\n",
      " |  \n",
      " |  normalize_schedule(self, dttm)\n",
      " |      Returns dttm + interval unless dttm is first interval then it returns dttm\n",
      " |  \n",
      " |  pickle(self, session=None)\n",
      " |  \n",
      " |  pickle_info(self, session=None)\n",
      " |  \n",
      " |  previous_schedule(self, dttm)\n",
      " |  \n",
      " |  resolve_template_files(self)\n",
      " |  \n",
      " |  run(self, start_date=None, end_date=None, mark_success=False, include_adhoc=False, local=False, executor=None, donot_pickle=False, ignore_task_deps=False, ignore_first_depends_on_past=False, pool=None)\n",
      " |      Runs the DAG.\n",
      " |  \n",
      " |  set_dag_runs_state(self, state='running', session=None)\n",
      " |  \n",
      " |  set_dependency(self, upstream_task_id, downstream_task_id)\n",
      " |      Simple utility method to set dependency between two tasks that\n",
      " |      already have been added to the DAG using add_task()\n",
      " |  \n",
      " |  sub_dag(self, task_regex, include_downstream=False, include_upstream=True)\n",
      " |      Returns a subset of the current dag as a deep copy of the current dag\n",
      " |      based on a regex that should match one or many tasks, and includes\n",
      " |      upstream and downstream neighbours based on the flag passed.\n",
      " |  \n",
      " |  topological_sort(self)\n",
      " |      Sorts tasks in topographical order, such that a task comes after any of its\n",
      " |      upstream dependencies.\n",
      " |      \n",
      " |      Heavily inspired by:\n",
      " |      http://blog.jupo.org/2012/04/06/topological-sorting-acyclic-directed-graphs/\n",
      " |      :returns: list of tasks in topological order\n",
      " |  \n",
      " |  tree_view(self)\n",
      " |      Shows an ascii tree representation of the DAG\n",
      " |  \n",
      " |  ----------------------------------------------------------------------\n",
      " |  Static methods defined here:\n",
      " |  \n",
      " |  deactivate_stale_dags(expiration_date, session=None)\n",
      " |      Deactivate any DAGs that were last touched by the scheduler before\n",
      " |      the expiration date. These DAGs were likely deleted.\n",
      " |      \n",
      " |      :param expiration_date: set inactive DAGs that were touched before this\n",
      " |      time\n",
      " |      :type expiration_date: datetime\n",
      " |      :return: None\n",
      " |  \n",
      " |  deactivate_unknown_dags(active_dag_ids, session=None)\n",
      " |      Given a list of known DAGs, deactivate any other DAGs that are\n",
      " |      marked as active in the ORM\n",
      " |      \n",
      " |      :param active_dag_ids: list of DAG IDs that are active\n",
      " |      :type active_dag_ids: list[unicode]\n",
      " |      :return: None\n",
      " |  \n",
      " |  sync_to_db(dag, owner, sync_time, session=None)\n",
      " |      Save attributes about this DAG to the DB. Note that this method\n",
      " |      can be called for both DAGs and SubDAGs. A SubDag is actually a\n",
      " |      SubDagOperator.\n",
      " |      \n",
      " |      :param dag: the DAG object to save to the DB\n",
      " |      :type dag: DAG\n",
      " |      :own\n",
      " |      :param sync_time: The time that the DAG should be marked as sync'ed\n",
      " |      :type sync_time: datetime\n",
      " |      :return: None\n",
      " |  \n",
      " |  ----------------------------------------------------------------------\n",
      " |  Data descriptors defined here:\n",
      " |  \n",
      " |  active_task_ids\n",
      " |  \n",
      " |  active_tasks\n",
      " |  \n",
      " |  concurrency\n",
      " |      :return: maximum number of tasks that can run simultaneously from this DAG\n",
      " |      :rtype: int\n",
      " |  \n",
      " |  concurrency_reached\n",
      " |      Returns a boolean indicating whether the concurrency limit for this DAG\n",
      " |      has been reached\n",
      " |  \n",
      " |  dag_id\n",
      " |      :return: the DAG ID\n",
      " |      :rtype: unicode\n",
      " |  \n",
      " |  description\n",
      " |  \n",
      " |  filepath\n",
      " |      File location of where the dag object is instantiated\n",
      " |  \n",
      " |  folder\n",
      " |      Folder location of where the dag object is instantiated\n",
      " |  \n",
      " |  full_filepath\n",
      " |      :return: The absolute path to the file that contains this DAG's definition\n",
      " |      :rtype: unicode\n",
      " |  \n",
      " |  is_paused\n",
      " |      Returns a boolean indicating whether this DAG is paused\n",
      " |  \n",
      " |  latest_execution_date\n",
      " |      Returns the latest date for which at least one dag run exists\n",
      " |  \n",
      " |  owner\n",
      " |  \n",
      " |  pickle_id\n",
      " |      :return: The pickle ID for this DAG, if it has one. Otherwise None.\n",
      " |      :rtype: unicode\n",
      " |  \n",
      " |  roots\n",
      " |  \n",
      " |  subdags\n",
      " |      Returns a list of the subdag objects associated to this DAG\n",
      " |  \n",
      " |  task_ids\n",
      " |      :return: A list of task IDs that are in this DAG\n",
      " |      :rtype: List[unicode]\n",
      " |  \n",
      " |  tasks\n",
      " |  \n",
      " |  ----------------------------------------------------------------------\n",
      " |  Data descriptors inherited from airflow.dag.base_dag.BaseDag:\n",
      " |  \n",
      " |  __dict__\n",
      " |      dictionary for instance variables (if defined)\n",
      " |  \n",
      " |  __weakref__\n",
      " |      list of weak references to the object (if defined)\n",
      " |  \n",
      " |  ----------------------------------------------------------------------\n",
      " |  Data and other attributes inherited from airflow.dag.base_dag.BaseDag:\n",
      " |  \n",
      " |  __metaclass__ = <class 'abc.ABCMeta'>\n",
      " |      Metaclass for defining Abstract Base Classes (ABCs).\n",
      " |      \n",
      " |      Use this metaclass to create an ABC.  An ABC can be subclassed\n",
      " |      directly, and then acts as a mix-in class.  You can also register\n",
      " |      unrelated concrete classes (even built-in classes) and unrelated\n",
      " |      ABCs as 'virtual subclasses' -- these and their descendants will\n",
      " |      be considered subclasses of the registering ABC by the built-in\n",
      " |      issubclass() function, but the registering ABC won't show up in\n",
      " |      their MRO (Method Resolution Order) nor will method\n",
      " |      implementations defined by the registering ABC be callable (not\n",
      " |      even via super()).\n",
      " |  \n",
      " |  ----------------------------------------------------------------------\n",
      " |  Data descriptors inherited from airflow.utils.logging.LoggingMixin:\n",
      " |  \n",
      " |  logger\n",
      "\n"
     ]
    }
   ],
   "source": [
    "help(DAG)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 定义DAG模型。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2018-07-21 14:23:30,273] {__init__.py:57} INFO - Using executor SequentialExecutor\n"
     ]
    }
   ],
   "source": [
    "\"\"\"\n",
    "Code that goes along with the Airflow tutorial located at:\n",
    "https://github.com/airbnb/airflow/blob/master/airflow/example_dags/tutorial.py\n",
    "\"\"\"\n",
    "from airflow import DAG\n",
    "from airflow.operators.bash_operator import BashOperator\n",
    "from datetime import datetime, timedelta\n",
    "\n",
    "\n",
    "default_args = {\n",
    "    'owner': 'airflow',\n",
    "    'depends_on_past': False,\n",
    "    'start_date': datetime(2015, 6, 1),\n",
    "    'email': ['airflow@example.com'],\n",
    "    'email_on_failure': False,\n",
    "    'email_on_retry': False,\n",
    "    'retries': 1,\n",
    "    'retry_delay': timedelta(minutes=5),\n",
    "    # 'queue': 'bash_queue',\n",
    "    # 'pool': 'backfill',\n",
    "    # 'priority_weight': 10,\n",
    "    # 'end_date': datetime(2016, 1, 1),\n",
    "}\n",
    "\n",
    "dag = DAG('tutorial', default_args=default_args)\n",
    "\n",
    "# t1, t2 and t3 are examples of tasks created by instantiating operators\n",
    "t1 = BashOperator(\n",
    "    task_id='print_date',\n",
    "    bash_command='date',\n",
    "    dag=dag)\n",
    "\n",
    "t2 = BashOperator(\n",
    "    task_id='sleep',\n",
    "    bash_command='sleep 5',\n",
    "    retries=3,\n",
    "    dag=dag)\n",
    "\n",
    "templated_command = \"\"\"\n",
    "    {% for i in range(5) %}\n",
    "        echo \"{{ ds }}\"\n",
    "        echo \"{{ macros.ds_add(ds, 7)}}\"\n",
    "        echo \"{{ params.my_param }}\"\n",
    "    {% endfor %}\n",
    "\"\"\"\n",
    "\n",
    "t3 = BashOperator(\n",
    "    task_id='templated',\n",
    "    bash_command=templated_command,\n",
    "    params={'my_param': 'Parameter I passed in'},\n",
    "    dag=dag)\n",
    "\n",
    "t2.set_upstream(t1)\n",
    "t3.set_upstream(t1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 查看定义信息。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[0;31mType:\u001b[0m        BashOperator\n",
       "\u001b[0;31mString form:\u001b[0m <Task(BashOperator): sleep>\n",
       "\u001b[0;31mFile:\u001b[0m        /srv/conda/lib/python3.6/site-packages/airflow/operators/bash_operator.py\n",
       "\u001b[0;31mDocstring:\u001b[0m  \n",
       "Execute a Bash script, command or set of commands.\n",
       "\n",
       ":param bash_command: The command, set of commands or reference to a\n",
       "    bash script (must be '.sh') to be executed.\n",
       ":type bash_command: string\n",
       ":param xcom_push: If xcom_push is True, the last line written to stdout\n",
       "    will also be pushed to an XCom when the bash command completes.\n",
       ":type xcom_push: bool\n",
       ":param env: If env is not None, it must be a mapping that defines the\n",
       "    environment variables for the new process; these are used instead\n",
       "    of inheriting the current process environment, which is the default\n",
       "    behavior. (templated)\n",
       ":type env: dict\n",
       ":type output_encoding: output encoding of bash command\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "?t2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[0;31mType:\u001b[0m        BashOperator\n",
       "\u001b[0;31mString form:\u001b[0m <Task(BashOperator): templated>\n",
       "\u001b[0;31mFile:\u001b[0m        /srv/conda/lib/python3.6/site-packages/airflow/operators/bash_operator.py\n",
       "\u001b[0;31mDocstring:\u001b[0m  \n",
       "Execute a Bash script, command or set of commands.\n",
       "\n",
       ":param bash_command: The command, set of commands or reference to a\n",
       "    bash script (must be '.sh') to be executed.\n",
       ":type bash_command: string\n",
       ":param xcom_push: If xcom_push is True, the last line written to stdout\n",
       "    will also be pushed to an XCom when the bash command completes.\n",
       ":type xcom_push: bool\n",
       ":param env: If env is not None, it must be a mapping that defines the\n",
       "    environment variables for the new process; these are used instead\n",
       "    of inheriting the current process environment, which is the default\n",
       "    behavior. (templated)\n",
       ":type env: dict\n",
       ":type output_encoding: output encoding of bash command\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "?t3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[0;31mType:\u001b[0m        BashOperator\n",
       "\u001b[0;31mString form:\u001b[0m <Task(BashOperator): print_date>\n",
       "\u001b[0;31mFile:\u001b[0m        /srv/conda/lib/python3.6/site-packages/airflow/operators/bash_operator.py\n",
       "\u001b[0;31mDocstring:\u001b[0m  \n",
       "Execute a Bash script, command or set of commands.\n",
       "\n",
       ":param bash_command: The command, set of commands or reference to a\n",
       "    bash script (must be '.sh') to be executed.\n",
       ":type bash_command: string\n",
       ":param xcom_push: If xcom_push is True, the last line written to stdout\n",
       "    will also be pushed to an XCom when the bash command completes.\n",
       ":type xcom_push: bool\n",
       ":param env: If env is not None, it must be a mapping that defines the\n",
       "    environment variables for the new process; these are used instead\n",
       "    of inheriting the current process environment, which is the default\n",
       "    behavior. (templated)\n",
       ":type env: dict\n",
       ":type output_encoding: output encoding of bash command\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "?t1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[<Task(BashOperator): print_date>,\n",
       " <Task(BashOperator): sleep>,\n",
       " <Task(BashOperator): templated>]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dag.active_tasks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<Task(BashOperator): sleep>\n",
      "    <Task(BashOperator): print_date>\n",
      "<Task(BashOperator): templated>\n",
      "    <Task(BashOperator): print_date>\n"
     ]
    }
   ],
   "source": [
    "dag.tree_view()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2018-07-21 14:23:50,730] {__init__.py:57} INFO - Using executor SequentialExecutor\n",
      "[2018-07-21 14:23:51,304] {models.py:167} INFO - Filling up the DagBag from /home/jovyan/airflow/dags\n",
      "Traceback (most recent call last):\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 1193, in _execute_context\n",
      "    context)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/default.py\", line 509, in do_execute\n",
      "    cursor.execute(statement, parameters)\n",
      "sqlite3.OperationalError: no such table: task_instance\n",
      "\n",
      "The above exception was the direct cause of the following exception:\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/srv/conda/bin/airflow\", line 28, in <module>\n",
      "    args.func(args)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/airflow/bin/cli.py\", line 585, in test\n",
      "    ti.run(ignore_task_deps=True, ignore_ti_state=True, test_mode=True)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/airflow/utils/db.py\", line 53, in wrapper\n",
      "    result = func(*args, **kwargs)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/airflow/models.py\", line 1249, in run\n",
      "    self.refresh_from_db(session=session, lock_for_update=True)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/airflow/utils/db.py\", line 53, in wrapper\n",
      "    result = func(*args, **kwargs)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/airflow/models.py\", line 991, in refresh_from_db\n",
      "    ti = qry.with_for_update().first()\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/query.py\", line 2888, in first\n",
      "    ret = list(self[0:1])\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/query.py\", line 2680, in __getitem__\n",
      "    return list(res)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/query.py\", line 2988, in __iter__\n",
      "    return self._execute_and_instances(context)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/query.py\", line 3011, in _execute_and_instances\n",
      "    result = conn.execute(querycontext.statement, self._params)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 948, in execute\n",
      "    return meth(self, multiparams, params)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/sql/elements.py\", line 269, in _execute_on_connection\n",
      "    return connection._execute_clauseelement(self, multiparams, params)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 1060, in _execute_clauseelement\n",
      "    compiled_sql, distilled_params\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 1200, in _execute_context\n",
      "    context)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 1413, in _handle_dbapi_exception\n",
      "    exc_info\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/util/compat.py\", line 265, in raise_from_cause\n",
      "    reraise(type(exception), exception, tb=exc_tb, cause=cause)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/util/compat.py\", line 248, in reraise\n",
      "    raise value.with_traceback(tb)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 1193, in _execute_context\n",
      "    context)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/default.py\", line 509, in do_execute\n",
      "    cursor.execute(statement, parameters)\n",
      "sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: task_instance [SQL: 'SELECT task_instance.task_id AS task_instance_task_id, task_instance.dag_id AS task_instance_dag_id, task_instance.execution_date AS task_instance_execution_date, task_instance.start_date AS task_instance_start_date, task_instance.end_date AS task_instance_end_date, task_instance.duration AS task_instance_duration, task_instance.state AS task_instance_state, task_instance.try_number AS task_instance_try_number, task_instance.hostname AS task_instance_hostname, task_instance.unixname AS task_instance_unixname, task_instance.job_id AS task_instance_job_id, task_instance.pool AS task_instance_pool, task_instance.queue AS task_instance_queue, task_instance.priority_weight AS task_instance_priority_weight, task_instance.operator AS task_instance_operator, task_instance.queued_dttm AS task_instance_queued_dttm, task_instance.pid AS task_instance_pid \\nFROM task_instance \\nWHERE task_instance.dag_id = ? AND task_instance.task_id = ? AND task_instance.execution_date = ?\\n LIMIT ? OFFSET ?'] [parameters: ('first_dag', 'print_date', '2015-06-01 00:00:00.000000', 1, 0)] (Background on this error at: http://sqlalche.me/e/e3q8)\n"
     ]
    }
   ],
   "source": [
    "!airflow test first_dag print_date 2015-06-01"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2018-07-21 12:03:01,781] {__init__.py:57} INFO - Using executor SequentialExecutor\n",
      "[2018-07-21 12:03:02,357] {models.py:167} INFO - Filling up the DagBag from /home/jovyan/airflow/dags\n",
      "Traceback (most recent call last):\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 1193, in _execute_context\n",
      "    context)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/default.py\", line 509, in do_execute\n",
      "    cursor.execute(statement, parameters)\n",
      "sqlite3.OperationalError: no such table: task_instance\n",
      "\n",
      "The above exception was the direct cause of the following exception:\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/srv/conda/bin/airflow\", line 28, in <module>\n",
      "    args.func(args)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/airflow/bin/cli.py\", line 585, in test\n",
      "    ti.run(ignore_task_deps=True, ignore_ti_state=True, test_mode=True)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/airflow/utils/db.py\", line 53, in wrapper\n",
      "    result = func(*args, **kwargs)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/airflow/models.py\", line 1249, in run\n",
      "    self.refresh_from_db(session=session, lock_for_update=True)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/airflow/utils/db.py\", line 53, in wrapper\n",
      "    result = func(*args, **kwargs)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/airflow/models.py\", line 991, in refresh_from_db\n",
      "    ti = qry.with_for_update().first()\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/query.py\", line 2888, in first\n",
      "    ret = list(self[0:1])\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/query.py\", line 2680, in __getitem__\n",
      "    return list(res)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/query.py\", line 2988, in __iter__\n",
      "    return self._execute_and_instances(context)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/query.py\", line 3011, in _execute_and_instances\n",
      "    result = conn.execute(querycontext.statement, self._params)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 948, in execute\n",
      "    return meth(self, multiparams, params)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/sql/elements.py\", line 269, in _execute_on_connection\n",
      "    return connection._execute_clauseelement(self, multiparams, params)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 1060, in _execute_clauseelement\n",
      "    compiled_sql, distilled_params\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 1200, in _execute_context\n",
      "    context)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 1413, in _handle_dbapi_exception\n",
      "    exc_info\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/util/compat.py\", line 265, in raise_from_cause\n",
      "    reraise(type(exception), exception, tb=exc_tb, cause=cause)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/util/compat.py\", line 248, in reraise\n",
      "    raise value.with_traceback(tb)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\", line 1193, in _execute_context\n",
      "    context)\n",
      "  File \"/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/default.py\", line 509, in do_execute\n",
      "    cursor.execute(statement, parameters)\n",
      "sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: task_instance [SQL: 'SELECT task_instance.task_id AS task_instance_task_id, task_instance.dag_id AS task_instance_dag_id, task_instance.execution_date AS task_instance_execution_date, task_instance.start_date AS task_instance_start_date, task_instance.end_date AS task_instance_end_date, task_instance.duration AS task_instance_duration, task_instance.state AS task_instance_state, task_instance.try_number AS task_instance_try_number, task_instance.hostname AS task_instance_hostname, task_instance.unixname AS task_instance_unixname, task_instance.job_id AS task_instance_job_id, task_instance.pool AS task_instance_pool, task_instance.queue AS task_instance_queue, task_instance.priority_weight AS task_instance_priority_weight, task_instance.operator AS task_instance_operator, task_instance.queued_dttm AS task_instance_queued_dttm, task_instance.pid AS task_instance_pid \\nFROM task_instance \\nWHERE task_instance.dag_id = ? AND task_instance.task_id = ? AND task_instance.execution_date = ?\\n LIMIT ? OFFSET ?'] [parameters: ('first_dag', 'sleep', '2015-06-01 00:00:00.000000', 1, 0)] (Background on this error at: http://sqlalche.me/e/e3q8)\n"
     ]
    }
   ],
   "source": [
    "!airflow test first_dag sleep 2015-06-01"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "ename": "OperationalError",
     "evalue": "(sqlite3.OperationalError) no such table: job [SQL: 'INSERT INTO job (dag_id, state, job_type, start_date, end_date, latest_heartbeat, executor_class, hostname, unixname) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)'] [parameters: ('tutorial', 'running', 'BackfillJob', '2018-07-21 11:57:02.407213', None, '2018-07-21 11:57:02.407221', 'SequentialExecutor', 'jupyter-openthings-2ddatabook-2d7grw66kv', 'jovyan')] (Background on this error at: http://sqlalche.me/e/e3q8)",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mOperationalError\u001b[0m                          Traceback (most recent call last)",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\u001b[0m in \u001b[0;36m_execute_context\u001b[0;34m(self, dialect, constructor, statement, parameters, *args)\u001b[0m\n\u001b[1;32m   1192\u001b[0m                         \u001b[0mparameters\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1193\u001b[0;31m                         context)\n\u001b[0m\u001b[1;32m   1194\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mBaseException\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/default.py\u001b[0m in \u001b[0;36mdo_execute\u001b[0;34m(self, cursor, statement, parameters, context)\u001b[0m\n\u001b[1;32m    508\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mdo_execute\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcursor\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mstatement\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mparameters\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcontext\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mNone\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 509\u001b[0;31m         \u001b[0mcursor\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexecute\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mstatement\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mparameters\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    510\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mOperationalError\u001b[0m: no such table: job",
      "\nThe above exception was the direct cause of the following exception:\n",
      "\u001b[0;31mOperationalError\u001b[0m                          Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-15-3f9cf16f40d1>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mdag\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrun\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/airflow/models.py\u001b[0m in \u001b[0;36mrun\u001b[0;34m(self, start_date, end_date, mark_success, include_adhoc, local, executor, donot_pickle, ignore_task_deps, ignore_first_depends_on_past, pool)\u001b[0m\n\u001b[1;32m   3328\u001b[0m             \u001b[0mignore_first_depends_on_past\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mignore_first_depends_on_past\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   3329\u001b[0m             pool=pool)\n\u001b[0;32m-> 3330\u001b[0;31m         \u001b[0mjob\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrun\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   3331\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   3332\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mcli\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/airflow/jobs.py\u001b[0m in \u001b[0;36mrun\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    192\u001b[0m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mstate\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mState\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mRUNNING\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    193\u001b[0m         \u001b[0msession\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0madd\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 194\u001b[0;31m         \u001b[0msession\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcommit\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    195\u001b[0m         \u001b[0mid_\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mid\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    196\u001b[0m         \u001b[0mmake_transient\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/session.py\u001b[0m in \u001b[0;36mcommit\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    941\u001b[0m                 \u001b[0;32mraise\u001b[0m \u001b[0msa_exc\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mInvalidRequestError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"No transaction is begun.\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    942\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 943\u001b[0;31m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtransaction\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcommit\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    944\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    945\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mprepare\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/session.py\u001b[0m in \u001b[0;36mcommit\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    465\u001b[0m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_assert_active\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mprepared_ok\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    466\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_state\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mPREPARED\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 467\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_prepare_impl\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    468\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    469\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_parent\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0;32mNone\u001b[0m \u001b[0;32mor\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mnested\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/session.py\u001b[0m in \u001b[0;36m_prepare_impl\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    445\u001b[0m                 \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msession\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_is_clean\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    446\u001b[0m                     \u001b[0;32mbreak\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 447\u001b[0;31m                 \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msession\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mflush\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    448\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    449\u001b[0m                 raise exc.FlushError(\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/session.py\u001b[0m in \u001b[0;36mflush\u001b[0;34m(self, objects)\u001b[0m\n\u001b[1;32m   2252\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   2253\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_flushing\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;32mTrue\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 2254\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_flush\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mobjects\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   2255\u001b[0m         \u001b[0;32mfinally\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   2256\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_flushing\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;32mFalse\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/session.py\u001b[0m in \u001b[0;36m_flush\u001b[0;34m(self, objects)\u001b[0m\n\u001b[1;32m   2378\u001b[0m         \u001b[0;32mexcept\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   2379\u001b[0m             \u001b[0;32mwith\u001b[0m \u001b[0mutil\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msafe_reraise\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 2380\u001b[0;31m                 \u001b[0mtransaction\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrollback\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0m_capture_exception\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   2381\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   2382\u001b[0m     def bulk_save_objects(\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py\u001b[0m in \u001b[0;36m__exit__\u001b[0;34m(self, type_, value, traceback)\u001b[0m\n\u001b[1;32m     64\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_exc_info\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;32mNone\u001b[0m   \u001b[0;31m# remove potential circular references\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     65\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwarn_only\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 66\u001b[0;31m                 \u001b[0mcompat\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mreraise\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mexc_type\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mexc_value\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mexc_tb\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     67\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     68\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mcompat\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mpy3k\u001b[0m \u001b[0;32mand\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_exc_info\u001b[0m \u001b[0;32mand\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_exc_info\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;36m1\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/util/compat.py\u001b[0m in \u001b[0;36mreraise\u001b[0;34m(tp, value, tb, cause)\u001b[0m\n\u001b[1;32m    247\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mvalue\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m__traceback__\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mtb\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    248\u001b[0m             \u001b[0;32mraise\u001b[0m \u001b[0mvalue\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwith_traceback\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtb\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 249\u001b[0;31m         \u001b[0;32mraise\u001b[0m \u001b[0mvalue\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    250\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    251\u001b[0m \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/session.py\u001b[0m in \u001b[0;36m_flush\u001b[0;34m(self, objects)\u001b[0m\n\u001b[1;32m   2342\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_warn_on_events\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;32mTrue\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   2343\u001b[0m             \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 2344\u001b[0;31m                 \u001b[0mflush_context\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexecute\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   2345\u001b[0m             \u001b[0;32mfinally\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   2346\u001b[0m                 \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_warn_on_events\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;32mFalse\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py\u001b[0m in \u001b[0;36mexecute\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    389\u001b[0m                     \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdependencies\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    390\u001b[0m                     postsort_actions):\n\u001b[0;32m--> 391\u001b[0;31m                 \u001b[0mrec\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexecute\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    392\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    393\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mfinalize_flush_changes\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py\u001b[0m in \u001b[0;36mexecute\u001b[0;34m(self, uow)\u001b[0m\n\u001b[1;32m    554\u001b[0m                              uow.states_for_mapper_hierarchy(\n\u001b[1;32m    555\u001b[0m                                  self.mapper, False, False),\n\u001b[0;32m--> 556\u001b[0;31m                              \u001b[0muow\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    557\u001b[0m                              )\n\u001b[1;32m    558\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py\u001b[0m in \u001b[0;36msave_obj\u001b[0;34m(base_mapper, states, uowtransaction, single)\u001b[0m\n\u001b[1;32m    179\u001b[0m         _emit_insert_statements(base_mapper, uowtransaction,\n\u001b[1;32m    180\u001b[0m                                 \u001b[0mcached_connections\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 181\u001b[0;31m                                 mapper, table, insert)\n\u001b[0m\u001b[1;32m    182\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    183\u001b[0m     _finalize_insert_update_commands(\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py\u001b[0m in \u001b[0;36m_emit_insert_statements\u001b[0;34m(base_mapper, uowtransaction, cached_connections, mapper, table, insert, bookkeeping)\u001b[0m\n\u001b[1;32m    864\u001b[0m                 \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    865\u001b[0m                     \u001b[0mresult\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mcached_connections\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0mconnection\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0;31m\\\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 866\u001b[0;31m                         \u001b[0mexecute\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mstatement\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mparams\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    867\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    868\u001b[0m                 \u001b[0mprimary_key\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mresult\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcontext\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0minserted_primary_key\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\u001b[0m in \u001b[0;36mexecute\u001b[0;34m(self, object, *multiparams, **params)\u001b[0m\n\u001b[1;32m    946\u001b[0m             \u001b[0;32mraise\u001b[0m \u001b[0mexc\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mObjectNotExecutableError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mobject\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    947\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 948\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mmeth\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mmultiparams\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mparams\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    949\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    950\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_execute_function\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mfunc\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mmultiparams\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mparams\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/sql/elements.py\u001b[0m in \u001b[0;36m_execute_on_connection\u001b[0;34m(self, connection, multiparams, params)\u001b[0m\n\u001b[1;32m    267\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_execute_on_connection\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mconnection\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mmultiparams\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mparams\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    268\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msupports_execution\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 269\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mconnection\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_execute_clauseelement\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mmultiparams\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mparams\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    270\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    271\u001b[0m             \u001b[0;32mraise\u001b[0m \u001b[0mexc\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mObjectNotExecutableError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\u001b[0m in \u001b[0;36m_execute_clauseelement\u001b[0;34m(self, elem, multiparams, params)\u001b[0m\n\u001b[1;32m   1058\u001b[0m             \u001b[0mcompiled_sql\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1059\u001b[0m             \u001b[0mdistilled_params\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1060\u001b[0;31m             \u001b[0mcompiled_sql\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdistilled_params\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1061\u001b[0m         )\n\u001b[1;32m   1062\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_has_events\u001b[0m \u001b[0;32mor\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mengine\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_has_events\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\u001b[0m in \u001b[0;36m_execute_context\u001b[0;34m(self, dialect, constructor, statement, parameters, *args)\u001b[0m\n\u001b[1;32m   1198\u001b[0m                 \u001b[0mparameters\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1199\u001b[0m                 \u001b[0mcursor\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1200\u001b[0;31m                 context)\n\u001b[0m\u001b[1;32m   1201\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1202\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_has_events\u001b[0m \u001b[0;32mor\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mengine\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_has_events\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\u001b[0m in \u001b[0;36m_handle_dbapi_exception\u001b[0;34m(self, e, statement, parameters, cursor, context)\u001b[0m\n\u001b[1;32m   1411\u001b[0m                 util.raise_from_cause(\n\u001b[1;32m   1412\u001b[0m                     \u001b[0msqlalchemy_exception\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1413\u001b[0;31m                     \u001b[0mexc_info\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1414\u001b[0m                 )\n\u001b[1;32m   1415\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/util/compat.py\u001b[0m in \u001b[0;36mraise_from_cause\u001b[0;34m(exception, exc_info)\u001b[0m\n\u001b[1;32m    263\u001b[0m     \u001b[0mexc_type\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mexc_value\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mexc_tb\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mexc_info\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    264\u001b[0m     \u001b[0mcause\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mexc_value\u001b[0m \u001b[0;32mif\u001b[0m \u001b[0mexc_value\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mexception\u001b[0m \u001b[0;32melse\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 265\u001b[0;31m     \u001b[0mreraise\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtype\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mexception\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mexception\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mtb\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mexc_tb\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcause\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mcause\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    266\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    267\u001b[0m \u001b[0;32mif\u001b[0m \u001b[0mpy3k\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/util/compat.py\u001b[0m in \u001b[0;36mreraise\u001b[0;34m(tp, value, tb, cause)\u001b[0m\n\u001b[1;32m    246\u001b[0m             \u001b[0mvalue\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m__cause__\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mcause\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    247\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mvalue\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m__traceback__\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mtb\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 248\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mvalue\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwith_traceback\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtb\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    249\u001b[0m         \u001b[0;32mraise\u001b[0m \u001b[0mvalue\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    250\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/base.py\u001b[0m in \u001b[0;36m_execute_context\u001b[0;34m(self, dialect, constructor, statement, parameters, *args)\u001b[0m\n\u001b[1;32m   1191\u001b[0m                         \u001b[0mstatement\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1192\u001b[0m                         \u001b[0mparameters\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1193\u001b[0;31m                         context)\n\u001b[0m\u001b[1;32m   1194\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mBaseException\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1195\u001b[0m             self._handle_dbapi_exception(\n",
      "\u001b[0;32m/srv/conda/lib/python3.6/site-packages/sqlalchemy/engine/default.py\u001b[0m in \u001b[0;36mdo_execute\u001b[0;34m(self, cursor, statement, parameters, context)\u001b[0m\n\u001b[1;32m    507\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    508\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mdo_execute\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcursor\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mstatement\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mparameters\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcontext\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mNone\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 509\u001b[0;31m         \u001b[0mcursor\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexecute\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mstatement\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mparameters\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    510\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    511\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mdo_execute_no_params\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcursor\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mstatement\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcontext\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mNone\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mOperationalError\u001b[0m: (sqlite3.OperationalError) no such table: job [SQL: 'INSERT INTO job (dag_id, state, job_type, start_date, end_date, latest_heartbeat, executor_class, hostname, unixname) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)'] [parameters: ('tutorial', 'running', 'BackfillJob', '2018-07-21 11:57:02.407213', None, '2018-07-21 11:57:02.407221', 'SequentialExecutor', 'jupyter-openthings-2ddatabook-2d7grw66kv', 'jovyan')] (Background on this error at: http://sqlalche.me/e/e3q8)"
     ]
    }
   ],
   "source": [
    "dag.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 运行DAG，查看信息。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2018-07-21 11:45:16,001] {__init__.py:57} INFO - Using executor SequentialExecutor\n"
     ]
    }
   ],
   "source": [
    "!python ~/airflow/dags/airflowfirst.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 查看DAG列表。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2018-07-21 11:45:41,296] {__init__.py:57} INFO - Using executor SequentialExecutor\n",
      "[2018-07-21 11:45:41,847] {models.py:167} INFO - Filling up the DagBag from /home/jovyan/airflow/dags\n",
      "\n",
      "\n",
      "-------------------------------------------------------------------\n",
      "DAGS\n",
      "-------------------------------------------------------------------\n",
      "example_bash_operator\n",
      "example_branch_dop_operator_v3\n",
      "example_branch_operator\n",
      "example_http_operator\n",
      "example_passing_params_via_test_command\n",
      "example_python_operator\n",
      "example_short_circuit_operator\n",
      "example_skip_dag\n",
      "example_subdag_operator\n",
      "example_subdag_operator.section-1\n",
      "example_subdag_operator.section-2\n",
      "example_trigger_controller_dag\n",
      "example_trigger_target_dag\n",
      "example_xcom\n",
      "first_dag\n",
      "latest_only\n",
      "latest_only_with_trigger\n",
      "test_utils\n",
      "tutorial\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!airflow list_dags"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 列出DAG中的任务。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2018-07-21 11:44:36,769] {__init__.py:57} INFO - Using executor SequentialExecutor\n",
      "[2018-07-21 11:44:37,318] {models.py:167} INFO - Filling up the DagBag from /home/jovyan/airflow/dags\n",
      "<Task(BashOperator): sleep>\n",
      "    <Task(BashOperator): print_date>\n",
      "<Task(BashOperator): templated>\n",
      "    <Task(BashOperator): print_date>\n"
     ]
    }
   ],
   "source": [
    "!airflow list_tasks first_dag --tree"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2018-07-21 11:47:35,807] {__init__.py:57} INFO - Using executor SequentialExecutor\n",
      "[2018-07-21 11:47:36,359] {models.py:167} INFO - Filling up the DagBag from /home/jovyan/airflow/dags\n",
      "<Task(BashOperator): sleep>\n",
      "    <Task(BashOperator): print_date>\n",
      "<Task(BashOperator): templated>\n",
      "    <Task(BashOperator): print_date>\n"
     ]
    }
   ],
   "source": [
    "!airflow list_tasks first_dag --tree"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
